package me.wang;

import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.util.UUID;

public class TestUnion {
    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        RichSourceFunction<String> richSinkFunction = new RichSourceFunction<String>() {
            //            private ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);

            @Override
            public void open(OpenContext openContext) throws Exception {

            }

            @Override
            public void run(SourceContext<String> sourceContext) throws Exception {
//                    executor.scheduleAtFixedRate(new DataGenerator(sourceContext), 0 , 1000L , TimeUnit.MILLISECONDS);
                while(true){
                    Thread.sleep(1000L);
                    sourceContext.collect(UUID.randomUUID().toString());
                }
            }

            @Override
            public void cancel() {

            }
        };
        DataStreamSource<String> source1 = env.addSource(richSinkFunction);
        DataStreamSource<String> source2 = env.addSource(richSinkFunction);
        source2.union(source2).map(x-> x+"_fake").print();
        env.execute();
    }
}
